package git.soulbgm.format;

import com.dyuproject.protostuff.ProtobufIOUtil;
import com.dyuproject.protostuff.Schema;
import com.dyuproject.protostuff.runtime.RuntimeSchema;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.types.RowKind;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * protostuff反序列化模式
 *
 * @author SoulBGM
 * @date 2024-09-18
 */
@Internal
public class ProtostuffDeserializationSchema implements DeserializationSchema<RowData> {

    private final TypeInformation<RowData> resultTypeInfo;

    public ProtostuffDeserializationSchema(TypeInformation<RowData> resultTypeInfo) {
        this.resultTypeInfo = resultTypeInfo;
    }

    @Override
    public RowData deserialize(byte[] message) throws IOException {
        UniversalPojo<?> pojo = deserializer(message, UniversalPojo.class);
        Object obj = pojo.getT();
        Field[] declaredFields = obj.getClass().getDeclaredFields();
        // 创建一个RowData实例，指定行的大小（字段数量）
        GenericRowData row = new GenericRowData(declaredFields.length);

        // 设置行的类型（操作类型，如INSERT）
        row.setRowKind(RowKind.INSERT);

        try {
            // 手动设置每个字段的值
            for (int i = 0; i < declaredFields.length; i++) {
                Field field = declaredFields[i];
                field.setAccessible(true);
                Object val = field.get(obj);
                if (field.getType() == Date.class) {
                    val = convertToDateData((Date) val);
                }
                row.setField(i, val);
            }
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }

        return row;
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return resultTypeInfo;
    }

    /**
     * 转换为日期数据
     *
     * @param date 日期
     * @return {@link TimestampData}
     */
    public static TimestampData convertToDateData(Date date) {
        return TimestampData.fromLocalDateTime(date.toInstant().atZone(java.time.ZoneId.systemDefault()).toLocalDateTime());
    }

    private static final Map<Class<?>, Schema<?>> CACHED_SCHEMA = new ConcurrentHashMap<>();

    private static <T> Schema<T> getSchema(Class<T> clazz) {
        @SuppressWarnings("unchecked")
        Schema<T> schema = (Schema<T>) CACHED_SCHEMA.get(clazz);
        if (schema == null) {
            schema = RuntimeSchema.createFrom(clazz);
            CACHED_SCHEMA.put(clazz, schema);
        }
        return schema;
    }

    /**
     * 将通过protostuff的方式序列化的byte[] 进行反序列化成实体对象
     *
     * @param data  protostuff的方式序列化的byte[]数据
     * @param clazz 要进行反序列化成实体的Class
     * @return {@link T}
     */
    public static <T> T deserializer(byte[] data, Class<T> clazz) {
        try {
            Schema<T> schema = getSchema(clazz);
            T t = schema.newMessage();
            ProtobufIOUtil.mergeFrom(data, t, schema);
            return t;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 通用实体
     *
     * @author SoulBGM
     * @date 2024/02/29
     */
    public static class UniversalPojo<T> {
        private T t;

        public UniversalPojo() {
        }

        public UniversalPojo(T t) {
            this.t = t;
        }

        public T getT() {
            return t;
        }
    }
}
